#include "StdAfx.h"
#include "ZIocp.h"
#include "ZIocpProcessThread.h"
#include "ZUtility.h"
#include <assert.h>

namespace	Z	{

//------------------------------------------------------------------------------------------------------------------------
Iocp::Iocp( )
:	mIocp(0)
{
}

Iocp::~Iocp( )
{
}

bool Iocp::create(DWORD numConcurrentThreads)
{
	mIocp	= ::CreateIoCompletionPort(INVALID_HANDLE_VALUE, 0, 0, numConcurrentThreads);
	assert(mIocp != 0);
	return mIocp != 0;
}

void Iocp::destroy()
{
	closeHandle(mIocp);
}

bool Iocp::associate(SOCKET so, ULONG_PTR completionKey)
{
	assert(so != INVALID_SOCKET);
	return associate((HANDLE)so, completionKey);
}

bool Iocp::associate(HANDLE hDevice, ULONG_PTR completionKey)
{
	assert(hDevice != INVALID_HANDLE_VALUE);
	assert(completionKey != 0);
	const HANDLE hIocp	= ::CreateIoCompletionPort(hDevice, mIocp, completionKey, 0);
	return hIocp == mIocp;
}

BOOL Iocp::postCompletionStatus(ULONG_PTR completionKey, DWORD dwBytesTransferred, LPOVERLAPPED pOverlapped)
{
	return ::PostQueuedCompletionStatus(mIocp, dwBytesTransferred, completionKey, pOverlapped);
}

BOOL Iocp::getCompletionStatus(PULONG_PTR pCompletionKey, LPDWORD pdwBytesTransferred, LPOVERLAPPED *ppOverlapped)
{
	return ::GetQueuedCompletionStatus(mIocp, pdwBytesTransferred, pCompletionKey, ppOverlapped, INFINITE); 
}

bool Iocp::createThreadPool(IocpProcessThread* pProcessThread, size_t numPooledThread)
{
	assert(0 <= numPooledThread);
	
	if(numPooledThread == 0)	{
		SystemInfo si;
		numPooledThread	= si.dwNumberOfProcessors * 2 + 2;
	}

	try	{
		assert(mThreadPool == 0);
		mThreadPool.reset(new boost::thread_group);
		for(size_t i = 0; i < numPooledThread; ++i)	{
			mThreadPool->create_thread(boost::bind(&IocpProcessThread::processingThread, pProcessThread));
		}
	}
	catch(const boost::thread_resource_error&)	{
		return false;
	}

	return true;
}

void Iocp::destroyThreadPool()
{
	if(mThreadPool != 0)	{
		for(size_t i = 0; i < mThreadPool->size(); ++i)	{
			postCompletionStatus(IocpProcessThread::destroy);
		}
		mThreadPool->join_all();
		mThreadPool.reset();
	}
}

}